#pragma once

#include "../mqcommon/log.hpp"
#include "../mqcommon/helper.hpp"
#include "../mqcommon/msg.pb.h"

#include <iostream>
#include <sstream>
#include <mutex>
#include <memory>
#include <unordered_map>
#include <list>

namespace mymq
{
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp"
    using MessagePtr = std::shared_ptr<mymq::Message>;

    class MessageMapper
    {
    public:
        MessageMapper(const std::string& basedir, const std::string& qname)
        : _datafile(basedir)
        , _tmpfile(basedir)
        {
            if (!FileHelper(basedir).Exists())
            {
                FileHelper::CreateDirectory(basedir);
            }
            if (basedir.back() != '/')
            {
                _datafile += '/';
                _tmpfile += '/';
            }
            _datafile += qname + DATAFILE_SUBFIX;
            _tmpfile += qname + TMPFILE_SUBFIX;
            CreateMsgFile();
        }
        void CreateMsgFile()
        {
            if (!FileHelper(_datafile).Exists())
            {
                bool ret = FileHelper::CreateFile(_datafile);
                if (ret == false)
                {
                    FATAL_LOG("创建原消息文件%s失败", _datafile.c_str());
                    abort();
                }
            }
        }
        void RemoveMsgFile()
        {
            FileHelper::RemoveFile(_datafile);
            FileHelper::RemoveFile(_tmpfile);
        }
        void Insert(MessagePtr& mp)
        {
            bool ret = Insert(_datafile, mp);
            if (ret == false)
            {
                ERROR_LOG("向原消息文件%s写入失败", _datafile.c_str());
            }
        }
        void Remove(MessagePtr& mp)
        {
            mp->mutable_payload()->set_valid("0");
            std::string msg_body = mp->payload().SerializeAsString();
            size_t msg_size = msg_body.size();
            if (msg_size != mp->len())
            {
                ERROR_LOG("不能修改原消息文件%s中的数据信息，因为新生成的数据与原数据长度不一致！", _datafile.c_str());
                return;
            }
            FileHelper fhelp(_datafile);
            fhelp.Write(msg_body.c_str(), mp->offset(), msg_size);
        }
        std::list<MessagePtr> Gc()
        {
            std::list<MessagePtr> result;
            bool ret = false;
            ret = Load(&result);
            if (ret == false)
            {
                ERROR_LOG("加载有效数据失败");
                return result;
            }
            if (!FileHelper(_tmpfile).Exists())
            {
                bool ret = FileHelper::CreateFile(_tmpfile);
                if (ret == false)
                {
                    FATAL_LOG("创建原消息文件%s失败", _tmpfile.c_str());
                    abort();
                }
            }
            for (auto& mp: result)
            {
                ret = Insert(_tmpfile, mp);
                if (ret == false)
                {
                    ERROR_LOG("向临时消息文件%s写入失败", _tmpfile.c_str());
                    return result;
                }
            }
            ret = FileHelper::RemoveFile(_datafile);
            if (ret == false)
            {
                ERROR_LOG("删除原消息文件%s失败", _datafile.c_str());
                FileHelper::RemoveFile(_tmpfile);  // 清理临时文件
                return result;
            }
            // 如果 _tmpfile不存在，Rename()会失败（返回false）
            ret = FileHelper(_tmpfile).Rename(_datafile);
            if (ret == false)
            {
                ERROR_LOG("修改临时消息文件%s为原消息文件%s失败", _tmpfile.c_str(), _datafile.c_str());
                return result;
            }
            return result;
        }
    private:
        bool Load(std::list<MessagePtr>* result)
        {
            // 4字节首部|消息大小|4字节首部|消息大小...
            bool ret = false;
            FileHelper fhelp(_datafile);
            size_t fsize = fhelp.Size();
            size_t offset = 0;
            INFO_LOG("准备加载原消息文件%s数据，当前文件大小：%ld", _datafile.c_str(), fsize);
            while (offset < fsize)
            {
                size_t msg_size = 0;
                ret = fhelp.Read((char*)&msg_size, offset, sizeof(size_t));
                if (ret == false)
                {
                    ERROR_LOG("读取消息长度失败");
                    return false;
                }
                std::string msg_body(msg_size, '\0');
                ret = fhelp.Read(&msg_body[0], offset + sizeof(size_t), msg_size);
                if (ret == false)
                {
                    ERROR_LOG("读取消息内容失败");
                    return false; 
                }
                offset += sizeof(size_t);
                MessagePtr mp = std::make_shared<Message>();
                mp->mutable_payload()->ParseFromString(msg_body);
                offset += msg_size;
                if (mp->payload().valid() == "0")
                {
                    INFO_LOG("加载到无效数据：%s，内容为%s", mp->payload().properties().id().c_str(), mp->payload().body().c_str());
                    continue;
                }
                INFO_LOG("加载到有效数据：%s，内容为%s", mp->payload().properties().id().c_str(), mp->payload().body().c_str());
                result->push_back(mp);
            }
            return true;
        }
        bool Insert(const std::string& filepath, MessagePtr& mp)
        {
            bool ret = false;
            FileHelper fhelp(filepath);
            size_t fsize = fhelp.Size();
            std::string msg_body = mp->payload().SerializeAsString();
            size_t msg_size = msg_body.size();
            ret = fhelp.Write((char*)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                ERROR_LOG("写入消息长度失败");
                return false;
            }
            ret = fhelp.Write(msg_body.c_str(), fsize + sizeof(size_t), msg_body.size());
            if (ret == false)
            {
                ERROR_LOG("写入消息内容失败");
                return false;
            }
            mp->set_offset(fsize + sizeof(size_t));
            mp->set_len(msg_size);
            return true;
        }
    private:
        std::string _datafile;
        std::string _tmpfile;
    };

    // 以队列为单位进行管理
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
    public:
        QueueMessage(const std::string& basedir, const std::string& qname)
        : _mapper(basedir, qname)
        , _valid_count(0)
        , _total_count(0)
        {
        }
        void Recovery()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _msgs = _mapper.Gc();
            for (const auto& msg: _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _valid_count = _total_count = _msgs.size();
        }
        void Insert(BasicProperties* bp, const std::string& body, bool queue_is_durable)
        {
            MessagePtr mp = std::make_shared<Message>();
            mp->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                mp->mutable_payload()->mutable_properties()->set_id(bp->id());
                mp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                mp->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                mp->mutable_payload()->mutable_properties()->set_id(UUIDHelper::Getuuid());
                mp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                mp->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            std::unique_lock<std::mutex> lock(_mtx);
            if (mp->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                mp->mutable_payload()->set_valid("1");
                _mapper.Insert(mp);
                _valid_count++;
                _total_count++;
                _durable_msgs.insert(std::make_pair(mp->payload().properties().id(), mp));
            }
            _msgs.push_back(mp);
        }
        // 每次删除消息后，判断是否需要垃圾回收
        void Remove(const std::string& msg_id)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                INFO_LOG("待确认消息表中没有你要删除的信息：%s", msg_id.c_str());
                return;
            }
            MessagePtr mp = it->second;
            if (mp->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                _mapper.Remove(mp);
                _durable_msgs.erase(msg_id);
                _valid_count--;
                Gc();
            }
            _waitack_msgs.erase(msg_id);
            INFO_LOG("%s消息已经确认，删除该待确认消息%s成功", msg_id.c_str(), it->second->payload().body().c_str());
        }
        MessagePtr Front()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            if (_msgs.empty()) return MessagePtr();
            MessagePtr mp = _msgs.front();
            _msgs.pop_front();
            _waitack_msgs.insert(std::make_pair(mp->payload().properties().id(), mp));
            return mp;
        }
        // 返回的是待推送消息队列（链表）中的消息个数
        size_t GetableCount()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _msgs.size();
        }
        size_t DurableCount()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _durable_msgs.size();
        }
        // 总的持久化消息数量
        size_t TotalCount()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _total_count;
        }
        size_t WaitackCount()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _waitack_msgs.size();
        }
        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _mapper.RemoveMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
        }
    private:
        bool CheckGc()
        {
            if (_total_count > 2000 && _valid_count * 2 < _total_count)
                return true;
            return false;
        }
        void Gc()
        {
            if (CheckGc() == false) return;
            std::list<mymq::MessagePtr> msgs = _mapper.Gc();
            for (const auto& msg: msgs)
            {
                std::string id = msg->payload().properties().id();

                // 只有需要持久化的消息才会考虑更新在文件中的偏移量和大小
                // // 更新待推送队列中的消息
                // for (auto& m : _msgs)
                // {
                //     if (m->payload().properties().id() == id)
                //     {
                //         m->set_offset(msg->offset());
                //         m->set_len(msg->len());
                //         break;
                //     }
                // }
                
                // // 更新待确认消息
                // auto waitack_it = _waitack_msgs.find(id);
                // if (waitack_it != _waitack_msgs.end())
                // {
                //     waitack_it->second->set_offset(msg->offset());
                //     waitack_it->second->set_len(msg->len());
                // }

                auto it = _durable_msgs.find(id);
                if (it == _durable_msgs.end())
                {
                    INFO_LOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理");
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(id, msg));
                    continue;
                }
                it->second->set_offset(msg->offset());
                it->second->set_len(msg->len());
            }
            // 更新当前的有效消息数量 & 总的持久化消息数量
            _valid_count = _total_count = msgs.size();
        }
    private:
        std::mutex _mtx;
        size_t _valid_count; // 待持久化文件中的有效消息数
        size_t _total_count; // 总的持久化消息数
        MessageMapper _mapper;
        std::list<MessagePtr> _msgs; // 待推送消息
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息hash
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息
    };

    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
    public:
        MessageManager(const std::string& basedir)
        : _basedir(basedir)
        {
        }
        void InitQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end()) return;
                qp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qp));
            }
            qp->Recovery();
        }
        void DestoryQueueMessage(const std::string& qname)
        {   
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) return;
                qp = it->second;
                _queue_msgs.erase(qname);
            }
            qp->Clear();
        }
        bool Insert(const std::string& qname, BasicProperties* bp, const std::string& body, bool queue_is_durable)
        {
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    INFO_LOG("向队列%s新增消息失败：没有找到消息管理句柄", qname.c_str());
                    return false;
                }
                qp = it->second;
            }
            qp->Insert(bp, body, queue_is_durable);
            return true;
        }
        MessagePtr Front(const std::string& qname)
        {
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    INFO_LOG("获取队列%s队首消息失败：没有找到消息管理句柄", qname.c_str());
                    return MessagePtr();
                }
                qp = it->second;
            }
            return qp->Front();
        }
        bool Ack(const std::string& qname, const std::string& msg_id)
        {
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    INFO_LOG("确认队列%s消息%s失败：没有找到消息管理句柄", qname.c_str(), msg_id.c_str());
                    return false;
                }
                qp = it->second;
            }
            qp->Remove(msg_id);
            return true;
        }
        size_t GetableCount(const std::string& qname)
        {
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    INFO_LOG("获取队列%s待推送消息数量失败：没有找到消息管理句柄", qname.c_str());
                    return 0;
                }
                qp = it->second;
            }
            return qp->GetableCount();
        }
        size_t TotalCount(const std::string& qname)
        {
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    INFO_LOG("获取队列%s总持久化消息数量失败：没有找到消息管理句柄", qname.c_str());
                    return 0;
                }
                qp = it->second;
            }
            return qp->TotalCount();
        }
        size_t DurableCount(const std::string& qname)
        {
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    INFO_LOG("获取队列%s有效持久化消息数量失败：没有找到消息管理句柄", qname.c_str());
                    return 0;
                }
                qp = it->second;
            }
            return qp->DurableCount();
        }
        size_t WaitackCount(const std::string& qname)
        {
            QueueMessage::ptr qp;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    INFO_LOG("获取队列%s待确认消息数量失败：没有找到消息管理句柄", qname.c_str());
                    return 0;
                }
                qp = it->second;
            }
            return qp->WaitackCount();
        }
    void Clear()
    {
        std::unique_lock<std::mutex> lock(_mtx);
        for (const auto& queue_msg: _queue_msgs)
        {
            queue_msg.second->Clear();
        }
    }
    private:
        std::mutex _mtx;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}